package com.star.spring.rocketmq.controller;


import java.time.Duration;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.apis.producer.Transaction;
import org.apache.rocketmq.client.common.Pair;
import org.apache.rocketmq.client.core.RocketMQClientTemplate;

import com.star.spring.rocketmq.model.message.UserMessage;
import com.star.spring.rocketmq.service.ClientProducerDemo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("/test")
public class TestController {

    @Autowired
    private ClientProducerDemo producerService;

    @Autowired
    private RocketMQClientTemplate rocketMQClientTemplate;

    @PostMapping("/send/all")
    public String sendMessages() {
        try {
            producerService.send();
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
        return "success";
    }

    /**
     * 发送普通message
     */
    @PostMapping("/send/normal")
    public void sendNormalMessage(String topic) {

        UserMessage userInfo = new UserMessage()
                .setId(1)
                .setUserName("name")
                .setUserAge((byte) 3);

        SendReceipt sendReceipt = rocketMQClientTemplate.syncSendNormalMessage(topic, userInfo);
        log.info("send normal message success, message id:{}", sendReceipt.getMessageId());

    }

    /**
     * 发送延时message
     * @param topic
     */
    @PostMapping("/send/delay")
    public void sendCustomMessage(String topic) {

        UserMessage userInfo = new UserMessage()
                .setId(1)
                .setUserName("name")
                .setUserAge((byte) 3);

        SendReceipt sendReceipt = rocketMQClientTemplate.syncSendDelayMessage(topic, userInfo, Duration.ofSeconds(45));
        log.info("send delay message success, message id:{}", sendReceipt.getMessageId());

    }

    /**
     * 发送顺序message
     * @param topic
     */
    @PostMapping("/send/fifo")
    public void sendFifo(String topic) {

        UserMessage userInfo = new UserMessage()
                .setId(1)
                .setUserName("name")
                .setUserAge((byte) 3);

        rocketMQClientTemplate.syncSendFifoMessage(topic, userInfo, "fifo-group");
    }

    @PostMapping("/send/transaction")
    public void sendTransactionMessage(String topic) throws ClientException {

        UserMessage userInfo = new UserMessage()
                .setId(1)
                .setUserName("name")
                .setUserAge((byte) 3);
        Message<UserMessage> message = MessageBuilder.withPayload(userInfo).build();

        Pair<SendReceipt, Transaction> pair =
                rocketMQClientTemplate.sendTransactionMessage(topic, message);

        SendReceipt sendReceipt = pair.getSendReceipt();
        log.info("transactionSend to topic {} sendReceipt={}", topic, sendReceipt);
        Transaction transaction = pair.getTransaction();
        // executed local transaction
        if (true) {
            transaction.commit();
        } else {
            transaction.rollback();
        }

    }
}
